package com.gtp.mq.demo.exchange_type.topic;

import java.io.IOException;

import org.apache.commons.lang.SerializationUtils;
import org.junit.Test;

import com.gtp.mq.demo.QueueBindFactory;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class TopicConsumer extends QueueBindFactory {

	@Test
	public void test1() throws Exception {

		/*
		 * 流量限制:
		 * int prefetchSize  ->rabbitmq好像未实现
		 * int prefetchCount ->qps一样的东西
		 * boolean global    ->是否全局(channel)
		 */
		channel.basicQos(0, 1, false);
				
		/***************************** 订阅消费 push模式 **********************/
		Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				try {

					Object msg = SerializationUtils.deserialize(body);
					System.out.println("Message Number " + msg);

					/*
					 * //非自动ack设置(应答) 
					 * channel.basicAck(envelope.getDeliveryTag(),false);
					 * 
					 * false/true:单条还是多条
					 * 
					 * note: 如果这时autoAck=true，那么当consumer接收到相关消息之后
					 * 还没来得及处理就crash掉了，那么这样也算数据丢失
					 */
					
					/*
					 * ##拒收 场景:当前消费者不能处理，可能需要重新安排并让其他消费者接收并处理它
					 * 拒绝接收,不处理,可以设置是否放回到队列中还是丢掉
					 * 
					 * basicReject 一次拒绝一条 
					 * #丢弃 channel.basicReject(deliveryTag,false); 
					 * #重新尝试 channel.basicReject(deliveryTag, true);
					 * 
					 * ！！！如果没有其他接收者监控这个queue的话，要注意一直无限循环发送的危险
					 * basicNack：可以一次拒绝N条消息
					 * channel.basicNack(deliveryTag,multiple, requeue);
					 * 
					 * #deliveryTag认作消息的偏移量
					 * 
					 * #若启用如上场景的处理，则建议在 finally中处理
					 */
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		};

		// 订阅 这里自动ack
		channel.basicConsume("mytopicQueue1", true, consumer);
		/***********************************************************/
		/*
		 * 消费方式2:拉取pull模式 GetResponse basicGet(String queue, boolean autoAck)
		 */

		/*
		 * #参数不一致 错误：inequivalent arg 'durable' for exchange 'csExchange' in
		 * vhost '/': received 使用不同的MQ客户端时，常常会出现以上错误信息。 如使用php作为product,使用java,
		 * rubby, python作为consume. 最常见的原因是: durable,
		 * auto_delete,passive参数不一致，保持参数一致性就ok了
		 */

		/*
		 * 封装建议 //状态提前定义好 enum Action { ACCEPT, // 处理成功 RETRY, // 可以重试的错误
		 * REJECT, // 无需重试的错误 }
		 * 
		 * 代码框架如下： Action action = Action.RETRY; try { //
		 * 如果成功完成则action=Action.ACCEPT // 让子类去处理
		 * 
		 * }catch (Exception e) { // 根据异常种类决定是ACCEPT、RETRY还是 REJECT }finally {
		 * // 通过finally块来保证Ack/Nack会且只会执行一次 if (action == Action.ACCEPT) {
		 * channel.basicAck(tag); } else if (action == Action.RETRY) {
		 * channel.basicNack(tag, false, true); } else { channel.basicNack(tag,
		 * false, false); } }
		 */
		System.in.read();
	}
}
